package com.edu.realtime.app.func;

import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.util.DimUtil;
import com.edu.realtime.util.PhoenixUtil;

/**
 * @author zcx
 * @create 2022-10-18 22:07 将流中的数据写到phoenix表中
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // 获取向phoenix中写入数据的表名
        String sinkTable = jsonObj.getString("sink_table");
        // 从jsonObj中过滤掉sink_table
        jsonObj.remove("sink_table");

        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        PhoenixUtil.executeSqlUsePrepareStatement(sinkTable, jsonObj);

        // 如果对维度表数据进行了更新，将Redis中缓存的数据删除掉
        if ("update".equals(type)) {
            DimUtil.deleteCached(sinkTable, jsonObj.getString("id"));
        }
    }
}
